/*


Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controllers

import (
	"context"
	"fmt"
	"reflect"
	"strconv"
	"sync"
	"time"

	"slime.io/slime/framework/util"

	cmap "github.com/orcaman/concurrent-map"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"

	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"
	"slime.io/slime/framework/bootstrap"
	slime_model "slime.io/slime/framework/model"
	"slime.io/slime/framework/model/metric"
	"slime.io/slime/framework/model/trigger"
	microservicev1alpha2 "slime.io/slime/modules/limiter/api/v1alpha2"
	"slime.io/slime/modules/limiter/model"
)

// SmartLimiterReconciler reconciles a SmartLimiter object
type SmartLimiterReconciler struct {
	client.Client                 // K8s的Client: Client knows how to perform CRUD operations on Kubernetes objects.
	Scheme        *runtime.Scheme // 这里应该是slime的group/version/kind

	cfg *microservicev1alpha2.Limiter
	env bootstrap.Environment
	// key is limiter's namespace and name
	// value is the host
	// use service.ns.svc.cluster.local as default if host not specified
	// or use mock host when limiter in gw and outbound
	interest cmap.ConcurrentMap // key 是 "namespace/name", value 是域名
	// reuse, or use another filed to store interested nn
	// key is the interested namespace/name
	// value is the metricInfo
	metricInfo   cmap.ConcurrentMap // key 是"namespace/name", value是slime_model.Endpoints
	MetricSource source.Source

	metricInfoLock sync.RWMutex

	lastUpdatePolicy     microservicev1alpha2.SmartLimiterSpec
	lastUpdatePolicyLock *sync.RWMutex

	watcherMetricChan <-chan metric.Metric
	tickerMetricChan  <-chan metric.Metric
}

// +kubebuilder:rbac:groups=microservice.slime.io,resources=smartlimiters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=microservice.slime.io,resources=smartlimiters/status,verbs=get;update;patch

func (r *SmartLimiterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log.Infof("begin reconcile, get smartlimiter %+v", req)
	instance := &microservicev1alpha2.SmartLimiter{}
	if err := r.Client.Get(ctx, req.NamespacedName, instance); err != nil {
		if errors.IsNotFound(err) {
			instance = nil
			err = nil
			log.Infof("smartlimiter %v not found", req.NamespacedName)
		} else {
			log.Errorf("get smartlimiter %v err, %s", req.NamespacedName, err)
			return reconcile.Result{}, err
		}
	}
	// deleted
	if instance == nil {
		r.lastUpdatePolicyLock.Lock()
		r.lastUpdatePolicy = microservicev1alpha2.SmartLimiterSpec{}
		r.lastUpdatePolicyLock.Unlock()
		r.RemoveInterested(req)
		log.Infof("%v is deleted", req)
		return reconcile.Result{}, nil
	} else {
		// add or update
		log.Infof("%v is added or updated", req)
		if !r.env.RevInScope(slime_model.IstioRevFromLabel(instance.Labels)) { // 指定namespace/name的SmartLimiter对象, 携带的Labels中label(istio.io/rev). 如果此label的值没有出现在配置中, 直接返回. 所以这里就是匹配以下下发下来的Request,是不是自己感兴趣的.
			log.Debugf("existing smartlimiter %v istiorev %s but our %s, skip ...",
				req.NamespacedName, slime_model.IstioRevFromLabel(instance.Labels), r.env.IstioRev())
			return ctrl.Result{}, nil
		}
		r.lastUpdatePolicyLock.RLock()
		if reflect.DeepEqual(instance.Spec, r.lastUpdatePolicy) { // 如果此request与之前的instance相同, 则直接返回.
			r.lastUpdatePolicyLock.RUnlock()
			return reconcile.Result{}, nil
		} else {
			if out, err := r.Validate(instance); err != nil { // 校验失败,直接返回.
				log.Errorf("invalid smartlimiter, %s", err)
				return reconcile.Result{}, nil
			} else {
				r.lastUpdatePolicyLock.RUnlock()
				r.lastUpdatePolicyLock.Lock()
				r.lastUpdatePolicy = instance.Spec // 更新成功
				r.lastUpdatePolicyLock.Unlock()
				r.RegisterInterest(instance, req, out) // 注册interest
			}
		}
	}
	return ctrl.Result{}, nil
}

func (r *SmartLimiterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&microservicev1alpha2.SmartLimiter{}).
		Complete(r)
}

// NewReconciler returns a new reconcile.Reconciler
func NewReconciler(opts ...ReconcilerOpts) *SmartLimiterReconciler {
	r := &SmartLimiterReconciler{
		metricInfo:           cmap.New(),
		interest:             cmap.New(),
		lastUpdatePolicyLock: &sync.RWMutex{},
	}
	for _, opt := range opts {
		opt(r)
	}
	return r
}

func (r *SmartLimiterReconciler) Clear() {
	r.interest.Clear()
	r.metricInfo.Clear()
}

func NewProducerConfig(env bootstrap.Environment, cfg *microservicev1alpha2.Limiter) (*metric.ProducerConfig, error) {
	pc := &metric.ProducerConfig{
		EnableWatcherProducer: false,
		WatcherProducerConfig: metric.WatcherProducerConfig{
			Name:       "smartLimiter-watcher",
			MetricChan: make(chan metric.Metric),
			WatcherTriggerConfig: trigger.WatcherTriggerConfig{
				Kinds: []schema.GroupVersionKind{
					{
						Group:   "",
						Version: "v1",
						Kind:    "Endpoints",
					},
				},
				EventChan:     make(chan trigger.WatcherEvent),
				DynamicClient: env.DynamicClient,
			},
		},
		EnableTickerProducer: true,
		TickerProducerConfig: metric.TickerProducerConfig{
			Name:       "smartLimiter-ticker",
			MetricChan: make(chan metric.Metric),
			TickerTriggerConfig: trigger.TickerTriggerConfig{
				Durations: []time.Duration{
					30 * time.Second,
				},
				EventChan: make(chan trigger.TickerEvent),
			},
		},
		StopChan: env.Stop,
	}

	if !cfg.GetDisableAdaptive() {
		log.Info("enable adaptive ratelimiter")
		prometheusSourceConfig, err := newPrometheusSourceConfig(env)
		if err != nil {
			return nil, err
		}
		log.Infof("create new prometheus client success")
		pc.PrometheusSourceConfig = prometheusSourceConfig
		pc.EnablePrometheusSource = true
		pc.EnableWatcherProducer = true
	} else {
		log.Info("disable adaptive ratelimiter and promql is closed")
		pc.EnableMockSource = true
	}
	return pc, nil
}

func (r *SmartLimiterReconciler) RegisterInterest(instance *microservicev1alpha2.SmartLimiter, req ctrl.Request, out bool) {
	key := FQN(req.Namespace, req.Name)

	// query hostname base on instance
	host := buildHost(instance, out)

	// set interest mapping
	r.interest.Set(key, host)
	log.Infof("set %s/%s = %s", instance.Namespace, instance.Name, host)
}

func (r *SmartLimiterReconciler) RemoveInterested(req ctrl.Request) {
	key := FQN(req.Namespace, req.Name)

	r.metricInfo.Pop(key) // 就是删除对应namespace/name
	r.interest.Pop(key)   // 就是删除对应namespace/name
	log.Infof("name %s, namespace %s has poped", req.Name, req.Namespace)
	// if contain global smart limiter, should delete info in configmap
	if !r.cfg.GetDisableGlobalRateLimit() {
		log.Infof("refresh global rate limiter configmap")
		refreshConfigMap([]*model.Descriptor{}, r, req.NamespacedName)
	} else {
		log.Info("global rate limiter is closed")
	}
}

func (r *SmartLimiterReconciler) Validate(instance *microservicev1alpha2.SmartLimiter) (bool, error) { //
	var out bool
	gw := instance.Spec.Gateway
	var target *microservicev1alpha2.Target

	for _, descriptors := range instance.Spec.Sets {
		for _, descriptor := range descriptors.Descriptor_ {

			if descriptor.Target == nil && instance.Spec.Target == nil {
				return out, fmt.Errorf("invalid target")
			}
			if descriptor.Action == nil {
				return out, fmt.Errorf("invalid action")
			}
			if descriptor.Condition == "" {
				return out, fmt.Errorf("invalid condition")
			}

			condition := descriptor.Condition
			direction := target.GetDirection()

			quota := descriptor.Action.Quota
			if gw || direction == model.Outbound {
				out = true
				if !r.cfg.GetDisableAdaptive() { // 如果开启自适应流控, 校验失败
					return out, fmt.Errorf("outbound/gw must disable adaptive limiter")
				}

				if condition != "true" {
					return out, fmt.Errorf("condition must true in outbound")
				}
				if _, err := strconv.Atoi(quota); err != nil {
					return out, fmt.Errorf("quota must be number in outbound")
				}
			}
		}
	}
	return out, nil
}

func FQN(ns, name string) string {
	return ns + "/" + name
}

func buildHost(instance *microservicev1alpha2.SmartLimiter, out bool) string { // 根据namespace和name, 构造域名.
	// get name.namespaces.svc.cluster.local by default
	host := util.UnityHost(instance.Name, instance.Namespace)

	if out {
		host = model.MockHost
	} else if instance.Spec.Host != "" {
		host = instance.Spec.Host
	}
	return host
}

type ReconcilerOpts func(reconciler *SmartLimiterReconciler)

func ReconcilerWithEnv(env bootstrap.Environment) ReconcilerOpts { // Option模式
	return func(sr *SmartLimiterReconciler) {
		sr.env = env
	}
}

func ReconcilerWithCfg(cfg *microservicev1alpha2.Limiter) ReconcilerOpts { // Option模式
	return func(sr *SmartLimiterReconciler) {
		sr.cfg = cfg
	}
}

func ReconcilerWithProducerConfig(pc *metric.ProducerConfig) ReconcilerOpts { // Option模式
	return func(sr *SmartLimiterReconciler) {
		sr.watcherMetricChan = pc.WatcherProducerConfig.MetricChan
		sr.tickerMetricChan = pc.TickerProducerConfig.MetricChan
		// reconciler defines producer metric handler
		pc.WatcherProducerConfig.NeedUpdateMetricHandler = sr.handleWatcherEvent
		pc.TickerProducerConfig.NeedUpdateMetricHandler = sr.handleTickerEvent
	}
}
